{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 6-7 使用spark-scala调用tensorflow2.0训练好的模型\n",
    "\n",
    "本篇文章介绍在spark中调用训练好的tensorflow模型进行预测的方法。\n",
    "\n",
    "本文内容的学习需要一定的spark和scala基础。\n",
    "\n",
    "如果使用pyspark的话会比较简单，只需要在每个excutor上用Python加载模型分别预测就可以了。\n",
    "\n",
    "但工程上为了性能考虑，通常使用的是scala版本的spark。\n",
    "\n",
    "本篇文章我们通过TensorFlow for Java 在spark中调用训练好的tensorflow模型。\n",
    "\n",
    "利用spark的分布式计算能力，从而可以让训练好的tensorflow模型在成百上千的机器上分布式并行执行模型推断。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 〇、spark-scala调用tensorflow模型概述"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "在spark(scala)中调用tensorflow模型进行预测需要完成以下几个步骤。\n",
    "\n",
    "（1）准备protobuf模型文件\n",
    "\n",
    "（2）创建spark(scala)项目，在项目中添加java版本的tensorflow对应的jar包依赖\n",
    "\n",
    "（3）在spark(scala)项目中driver端加载tensorflow模型调试成功\n",
    "\n",
    "（4）在spark(scala)项目中通过RDD在excutor上加载tensorflow模型调试成功\n",
    "\n",
    "（5）在spark(scala)项目中通过DataFrame在excutor上加载tensorflow模型调试成功"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 一、准备protobuf模型文件\n",
    "\n",
    "我们使用tf.keras 训练一个简单的线性回归模型，并保存成protobuf文件。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Model: \"model\"\n",
      "_________________________________________________________________\n",
      "Layer (type)                 Output Shape              Param #   \n",
      "=================================================================\n",
      "inputs (InputLayer)          [(None, 2)]               0         \n",
      "_________________________________________________________________\n",
      "outputs (Dense)              (None, 1)                 3         \n",
      "=================================================================\n",
      "Total params: 3\n",
      "Trainable params: 3\n",
      "Non-trainable params: 0\n",
      "_________________________________________________________________\n",
      "Train on 800 samples\n",
      "Epoch 1/100\n",
      "800/800 [==============================] - 0s 290us/sample - loss: 329.1430 - mae: 14.9453\n",
      "Epoch 2/100\n",
      "800/800 [==============================] - 0s 98us/sample - loss: 303.8100 - mae: 14.3592\n",
      "Epoch 3/100\n",
      "800/800 [==============================] - 0s 103us/sample - loss: 278.8892 - mae: 13.7638\n",
      "Epoch 4/100\n",
      "800/800 [==============================] - 0s 99us/sample - loss: 255.2491 - mae: 13.1729\n",
      "Epoch 5/100\n",
      "800/800 [==============================] - 0s 87us/sample - loss: 232.9349 - mae: 12.5859\n",
      "Epoch 6/100\n",
      "800/800 [==============================] - 0s 94us/sample - loss: 212.1357 - mae: 12.0081\n",
      "Epoch 7/100\n",
      "800/800 [==============================] - 0s 91us/sample - loss: 191.8853 - mae: 11.4274\n",
      "Epoch 8/100\n",
      "800/800 [==============================] - 0s 93us/sample - loss: 172.7393 - mae: 10.8447\n",
      "Epoch 9/100\n",
      "800/800 [==============================] - 0s 91us/sample - loss: 154.8916 - mae: 10.2704\n",
      "Epoch 10/100\n",
      "800/800 [==============================] - 0s 94us/sample - loss: 138.5416 - mae: 9.7074\n",
      "Epoch 11/100\n",
      "800/800 [==============================] - 0s 78us/sample - loss: 122.4296 - mae: 9.1367\n",
      "Epoch 12/100\n",
      "800/800 [==============================] - 0s 86us/sample - loss: 108.1728 - mae: 8.5844\n",
      "Epoch 13/100\n",
      "800/800 [==============================] - 0s 91us/sample - loss: 94.4558 - mae: 8.0219\n",
      "Epoch 14/100\n",
      "800/800 [==============================] - 0s 91us/sample - loss: 81.5468 - mae: 7.4581\n",
      "Epoch 15/100\n",
      "800/800 [==============================] - 0s 91us/sample - loss: 69.8449 - mae: 6.9054\n",
      "Epoch 16/100\n",
      "800/800 [==============================] - 0s 90us/sample - loss: 59.4308 - mae: 6.3690\n",
      "Epoch 17/100\n",
      "800/800 [==============================] - 0s 88us/sample - loss: 49.9037 - mae: 5.8350\n",
      "Epoch 18/100\n",
      "800/800 [==============================] - 0s 89us/sample - loss: 41.3877 - mae: 5.3162\n",
      "Epoch 19/100\n",
      "800/800 [==============================] - 0s 89us/sample - loss: 33.9693 - mae: 4.8155\n",
      "Epoch 20/100\n",
      "800/800 [==============================] - 0s 90us/sample - loss: 27.4229 - mae: 4.3170\n",
      "Epoch 21/100\n",
      "800/800 [==============================] - 0s 91us/sample - loss: 21.7809 - mae: 3.8363\n",
      "Epoch 22/100\n",
      "800/800 [==============================] - 0s 90us/sample - loss: 17.1158 - mae: 3.3852\n",
      "Epoch 23/100\n",
      "800/800 [==============================] - 0s 104us/sample - loss: 13.5018 - mae: 2.9926\n",
      "Epoch 24/100\n",
      "800/800 [==============================] - 0s 100us/sample - loss: 10.6429 - mae: 2.6514\n",
      "Epoch 25/100\n",
      "800/800 [==============================] - 0s 95us/sample - loss: 8.5096 - mae: 2.3741\n",
      "Epoch 26/100\n",
      "800/800 [==============================] - 0s 90us/sample - loss: 7.1034 - mae: 2.1731\n",
      "Epoch 27/100\n",
      "800/800 [==============================] - 0s 109us/sample - loss: 6.2669 - mae: 2.0386\n",
      "Epoch 28/100\n",
      "800/800 [==============================] - 0s 102us/sample - loss: 5.8134 - mae: 1.9544\n",
      "Epoch 29/100\n",
      "800/800 [==============================] - 0s 90us/sample - loss: 5.5347 - mae: 1.8968\n",
      "Epoch 30/100\n",
      "800/800 [==============================] - 0s 91us/sample - loss: 5.3326 - mae: 1.8561\n",
      "Epoch 31/100\n",
      "800/800 [==============================] - 0s 87us/sample - loss: 5.1577 - mae: 1.8176\n",
      "Epoch 32/100\n",
      "800/800 [==============================] - 0s 90us/sample - loss: 5.0001 - mae: 1.7861\n",
      "Epoch 33/100\n",
      "800/800 [==============================] - 0s 85us/sample - loss: 4.8637 - mae: 1.7570\n",
      "Epoch 34/100\n",
      "800/800 [==============================] - 0s 86us/sample - loss: 4.7484 - mae: 1.7352\n",
      "Epoch 35/100\n",
      "800/800 [==============================] - 0s 93us/sample - loss: 4.6511 - mae: 1.7149\n",
      "Epoch 36/100\n",
      "800/800 [==============================] - 0s 90us/sample - loss: 4.5627 - mae: 1.6963\n",
      "Epoch 37/100\n",
      "800/800 [==============================] - 0s 90us/sample - loss: 4.4902 - mae: 1.6816\n",
      "Epoch 38/100\n",
      "800/800 [==============================] - 0s 90us/sample - loss: 4.4284 - mae: 1.6688\n",
      "Epoch 39/100\n",
      "800/800 [==============================] - 0s 84us/sample - loss: 4.3738 - mae: 1.6594\n",
      "Epoch 40/100\n",
      "800/800 [==============================] - 0s 82us/sample - loss: 4.3324 - mae: 1.6500\n",
      "Epoch 41/100\n",
      "800/800 [==============================] - 0s 80us/sample - loss: 4.2950 - mae: 1.6409\n",
      "Epoch 42/100\n",
      "800/800 [==============================] - 0s 84us/sample - loss: 4.2663 - mae: 1.6369\n",
      "Epoch 43/100\n",
      "800/800 [==============================] - 0s 82us/sample - loss: 4.2450 - mae: 1.6326\n",
      "Epoch 44/100\n",
      "800/800 [==============================] - 0s 80us/sample - loss: 4.2262 - mae: 1.6289\n",
      "Epoch 45/100\n",
      "800/800 [==============================] - 0s 82us/sample - loss: 4.2136 - mae: 1.6253\n",
      "Epoch 46/100\n",
      "800/800 [==============================] - 0s 79us/sample - loss: 4.2001 - mae: 1.6231\n",
      "Epoch 47/100\n",
      "800/800 [==============================] - 0s 78us/sample - loss: 4.1936 - mae: 1.6223\n",
      "Epoch 48/100\n",
      "800/800 [==============================] - 0s 78us/sample - loss: 4.1876 - mae: 1.6221\n",
      "Epoch 49/100\n",
      "800/800 [==============================] - 0s 79us/sample - loss: 4.1806 - mae: 1.6192\n",
      "Epoch 50/100\n",
      "800/800 [==============================] - 0s 90us/sample - loss: 4.1772 - mae: 1.6185\n",
      "Epoch 51/100\n",
      "800/800 [==============================] - 0s 84us/sample - loss: 4.1719 - mae: 1.6178\n",
      "Epoch 52/100\n",
      "800/800 [==============================] - 0s 83us/sample - loss: 4.1709 - mae: 1.6188\n",
      "Epoch 53/100\n",
      "800/800 [==============================] - 0s 85us/sample - loss: 4.1648 - mae: 1.6166\n",
      "Epoch 54/100\n",
      "800/800 [==============================] - 0s 92us/sample - loss: 4.1695 - mae: 1.6197\n",
      "Epoch 55/100\n",
      "800/800 [==============================] - 0s 89us/sample - loss: 4.1691 - mae: 1.6185\n",
      "Epoch 56/100\n",
      "800/800 [==============================] - 0s 87us/sample - loss: 4.1639 - mae: 1.6176\n",
      "Epoch 57/100\n",
      "800/800 [==============================] - 0s 91us/sample - loss: 4.1663 - mae: 1.6189\n",
      "Epoch 58/100\n",
      "800/800 [==============================] - 0s 89us/sample - loss: 4.1665 - mae: 1.6182\n",
      "Epoch 59/100\n",
      "800/800 [==============================] - 0s 80us/sample - loss: 4.1653 - mae: 1.6178\n",
      "Epoch 60/100\n",
      "800/800 [==============================] - 0s 80us/sample - loss: 4.1656 - mae: 1.6185\n",
      "Epoch 61/100\n",
      "800/800 [==============================] - 0s 79us/sample - loss: 4.1644 - mae: 1.6172\n",
      "Epoch 62/100\n",
      "800/800 [==============================] - 0s 79us/sample - loss: 4.1650 - mae: 1.6180\n",
      "Epoch 63/100\n",
      "800/800 [==============================] - 0s 81us/sample - loss: 4.1631 - mae: 1.6177\n",
      "Epoch 64/100\n",
      "800/800 [==============================] - 0s 85us/sample - loss: 4.1646 - mae: 1.6183\n",
      "Epoch 65/100\n",
      "800/800 [==============================] - 0s 84us/sample - loss: 4.1652 - mae: 1.6182\n",
      "Epoch 66/100\n",
      "800/800 [==============================] - 0s 83us/sample - loss: 4.1653 - mae: 1.6178\n",
      "Epoch 67/100\n",
      "800/800 [==============================] - 0s 82us/sample - loss: 4.1662 - mae: 1.6176\n",
      "Epoch 68/100\n",
      "800/800 [==============================] - 0s 88us/sample - loss: 4.1644 - mae: 1.6176\n",
      "Epoch 69/100\n",
      "800/800 [==============================] - 0s 89us/sample - loss: 4.1647 - mae: 1.6178\n",
      "Epoch 70/100\n",
      "800/800 [==============================] - 0s 88us/sample - loss: 4.1645 - mae: 1.6188\n",
      "Epoch 71/100\n",
      "800/800 [==============================] - 0s 89us/sample - loss: 4.1624 - mae: 1.6174\n",
      "Epoch 72/100\n",
      "800/800 [==============================] - 0s 87us/sample - loss: 4.1642 - mae: 1.6185\n",
      "Epoch 73/100\n",
      "800/800 [==============================] - 0s 85us/sample - loss: 4.1635 - mae: 1.6182\n",
      "Epoch 74/100\n",
      "800/800 [==============================] - 0s 90us/sample - loss: 4.1666 - mae: 1.6184\n",
      "Epoch 75/100\n",
      "800/800 [==============================] - 0s 89us/sample - loss: 4.1621 - mae: 1.6179\n",
      "Epoch 76/100\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "800/800 [==============================] - 0s 91us/sample - loss: 4.1649 - mae: 1.6177\n",
      "Epoch 77/100\n",
      "800/800 [==============================] - 0s 88us/sample - loss: 4.1645 - mae: 1.6181\n",
      "Epoch 78/100\n",
      "800/800 [==============================] - 0s 86us/sample - loss: 4.1648 - mae: 1.6175\n",
      "Epoch 79/100\n",
      "800/800 [==============================] - 0s 88us/sample - loss: 4.1653 - mae: 1.6180\n",
      "Epoch 80/100\n",
      "800/800 [==============================] - 0s 90us/sample - loss: 4.1643 - mae: 1.6180\n",
      "Epoch 81/100\n",
      "800/800 [==============================] - 0s 88us/sample - loss: 4.1639 - mae: 1.6181\n",
      "Epoch 82/100\n",
      "800/800 [==============================] - 0s 85us/sample - loss: 4.1631 - mae: 1.6182\n",
      "Epoch 83/100\n",
      "800/800 [==============================] - 0s 89us/sample - loss: 4.1657 - mae: 1.6191\n",
      "Epoch 84/100\n",
      "800/800 [==============================] - 0s 85us/sample - loss: 4.1648 - mae: 1.6180\n",
      "Epoch 85/100\n",
      "800/800 [==============================] - 0s 81us/sample - loss: 4.1642 - mae: 1.6179\n",
      "Epoch 86/100\n",
      "800/800 [==============================] - 0s 80us/sample - loss: 4.1652 - mae: 1.6180\n",
      "Epoch 87/100\n",
      "800/800 [==============================] - 0s 79us/sample - loss: 4.1649 - mae: 1.6173\n",
      "Epoch 88/100\n",
      "800/800 [==============================] - 0s 77us/sample - loss: 4.1636 - mae: 1.6179\n",
      "Epoch 89/100\n",
      "800/800 [==============================] - 0s 85us/sample - loss: 4.1633 - mae: 1.6185\n",
      "Epoch 90/100\n",
      "800/800 [==============================] - 0s 81us/sample - loss: 4.1629 - mae: 1.6163\n",
      "Epoch 91/100\n",
      "800/800 [==============================] - 0s 79us/sample - loss: 4.1644 - mae: 1.6186\n",
      "Epoch 92/100\n",
      "800/800 [==============================] - 0s 80us/sample - loss: 4.1629 - mae: 1.6171\n",
      "Epoch 93/100\n",
      "800/800 [==============================] - 0s 81us/sample - loss: 4.1647 - mae: 1.6179\n",
      "Epoch 94/100\n",
      "800/800 [==============================] - 0s 77us/sample - loss: 4.1646 - mae: 1.6179\n",
      "Epoch 95/100\n",
      "800/800 [==============================] - 0s 83us/sample - loss: 4.1637 - mae: 1.6175\n",
      "Epoch 96/100\n",
      "800/800 [==============================] - 0s 87us/sample - loss: 4.1645 - mae: 1.6179\n",
      "Epoch 97/100\n",
      "800/800 [==============================] - 0s 87us/sample - loss: 4.1644 - mae: 1.6185\n",
      "Epoch 98/100\n",
      "800/800 [==============================] - 0s 85us/sample - loss: 4.1645 - mae: 1.6180\n",
      "Epoch 99/100\n",
      "800/800 [==============================] - 0s 90us/sample - loss: 4.1659 - mae: 1.6179\n",
      "Epoch 100/100\n",
      "800/800 [==============================] - 0s 89us/sample - loss: 4.1644 - mae: 1.6183\n",
      "w =  [[1.97630334]\n",
      " [-1.00233161]]\n",
      "b =  [3.04598141]\n",
      "WARNING:tensorflow:From /Users/alan/anaconda3/lib/python3.7/site-packages/tensorflow_core/python/ops/resource_variable_ops.py:1786: calling BaseResourceVariable.__init__ (from tensorflow.python.ops.resource_variable_ops) with constraint is deprecated and will be removed in a future version.\n",
      "Instructions for updating:\n",
      "If using Keras pass *_constraint arguments to layers.\n",
      "INFO:tensorflow:Assets written to: ./data/linear_model/1/assets\n"
     ]
    }
   ],
   "source": [
    "import tensorflow as tf\n",
    "from tensorflow.keras import models,layers,optimizers\n",
    "\n",
    "## 样本数量\n",
    "n = 800\n",
    "\n",
    "## 生成测试用数据集\n",
    "X = tf.random.uniform([n,2],minval=-10,maxval=10) \n",
    "w0 = tf.constant([[2.0],[-1.0]])\n",
    "b0 = tf.constant(3.0)\n",
    "\n",
    "Y = X@w0 + b0 + tf.random.normal([n,1],mean = 0.0,stddev= 2.0)  # @表示矩阵乘法,增加正态扰动\n",
    "\n",
    "## 建立模型\n",
    "tf.keras.backend.clear_session()\n",
    "inputs = layers.Input(shape = (2,),name =\"inputs\") #设置输入名字为inputs\n",
    "outputs = layers.Dense(1, name = \"outputs\")(inputs) #设置输出名字为outputs\n",
    "linear = models.Model(inputs = inputs,outputs = outputs)\n",
    "linear.summary()\n",
    "\n",
    "## 使用fit方法进行训练\n",
    "linear.compile(optimizer=\"rmsprop\",loss=\"mse\",metrics=[\"mae\"])\n",
    "linear.fit(X,Y,batch_size = 8,epochs = 100)  \n",
    "\n",
    "tf.print(\"w = \",linear.layers[1].kernel)\n",
    "tf.print(\"b = \",linear.layers[1].bias)\n",
    "\n",
    "## 将模型保存成pb格式文件\n",
    "export_path = \"./data/linear_model/\"\n",
    "version = \"1\"       #后续可以通过版本号进行模型版本迭代与管理\n",
    "linear.save(export_path+version, save_format=\"tf\") \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\u001b[1m\u001b[36massets\u001b[m\u001b[m         saved_model.pb \u001b[1m\u001b[36mvariables\u001b[m\u001b[m\r\n"
     ]
    }
   ],
   "source": [
    "!ls {export_path+version}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "MetaGraphDef with tag-set: 'serve' contains the following SignatureDefs:\n",
      "\n",
      "signature_def['__saved_model_init_op']:\n",
      "  The given SavedModel SignatureDef contains the following input(s):\n",
      "  The given SavedModel SignatureDef contains the following output(s):\n",
      "    outputs['__saved_model_init_op'] tensor_info:\n",
      "        dtype: DT_INVALID\n",
      "        shape: unknown_rank\n",
      "        name: NoOp\n",
      "  Method name is: \n",
      "\n",
      "signature_def['serving_default']:\n",
      "  The given SavedModel SignatureDef contains the following input(s):\n",
      "    inputs['inputs'] tensor_info:\n",
      "        dtype: DT_FLOAT\n",
      "        shape: (-1, 2)\n",
      "        name: serving_default_inputs:0\n",
      "  The given SavedModel SignatureDef contains the following output(s):\n",
      "    outputs['outputs'] tensor_info:\n",
      "        dtype: DT_FLOAT\n",
      "        shape: (-1, 1)\n",
      "        name: StatefulPartitionedCall:0\n",
      "  Method name is: tensorflow/serving/predict\n",
      "WARNING:tensorflow:From /Users/alan/anaconda3/lib/python3.7/site-packages/tensorflow_core/python/ops/resource_variable_ops.py:1786: calling BaseResourceVariable.__init__ (from tensorflow.python.ops.resource_variable_ops) with constraint is deprecated and will be removed in a future version.\n",
      "Instructions for updating:\n",
      "If using Keras pass *_constraint arguments to layers.\n",
      "\n",
      "Defined Functions:\n",
      "  Function Name: '__call__'\n",
      "    Option #1\n",
      "      Callable with:\n",
      "        Argument #1\n",
      "          inputs: TensorSpec(shape=(None, 2), dtype=tf.float32, name='inputs')\n",
      "        Argument #2\n",
      "          DType: bool\n",
      "          Value: True\n",
      "        Argument #3\n",
      "          DType: NoneType\n",
      "          Value: None\n",
      "    Option #2\n",
      "      Callable with:\n",
      "        Argument #1\n",
      "          inputs: TensorSpec(shape=(None, 2), dtype=tf.float32, name='inputs')\n",
      "        Argument #2\n",
      "          DType: bool\n",
      "          Value: False\n",
      "        Argument #3\n",
      "          DType: NoneType\n",
      "          Value: None\n",
      "\n",
      "  Function Name: '_default_save_signature'\n",
      "    Option #1\n",
      "      Callable with:\n",
      "        Argument #1\n",
      "          inputs: TensorSpec(shape=(None, 2), dtype=tf.float32, name='inputs')\n",
      "\n",
      "  Function Name: 'call_and_return_all_conditional_losses'\n",
      "    Option #1\n",
      "      Callable with:\n",
      "        Argument #1\n",
      "          inputs: TensorSpec(shape=(None, 2), dtype=tf.float32, name='inputs')\n",
      "        Argument #2\n",
      "          DType: bool\n",
      "          Value: True\n",
      "        Argument #3\n",
      "          DType: NoneType\n",
      "          Value: None\n",
      "    Option #2\n",
      "      Callable with:\n",
      "        Argument #1\n",
      "          inputs: TensorSpec(shape=(None, 2), dtype=tf.float32, name='inputs')\n",
      "        Argument #2\n",
      "          DType: bool\n",
      "          Value: False\n",
      "        Argument #3\n",
      "          DType: NoneType\n",
      "          Value: None\n"
     ]
    }
   ],
   "source": [
    "# 查看模型文件相关信息\n",
    "!saved_model_cli show --dir {export_path+str(version)} --all"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "模型文件信息中这些标红的部分都是后面有可能会用到的。\n",
    "\n",
    "![](./data/模型文件信息.png)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 二、创建spark(scala)项目，在项目中添加java版本的tensorflow对应的jar包依赖\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "如果使用maven管理项目，需要添加如下 jar包依赖\n",
    "\n",
    "```\n",
    "<!-- https://mvnrepository.com/artifact/org.tensorflow/tensorflow -->\n",
    "<dependency>\n",
    "    <groupId>org.tensorflow</groupId>\n",
    "    <artifactId>tensorflow</artifactId>\n",
    "    <version>1.15.0</version>\n",
    "</dependency>\n",
    "```\n",
    "\n",
    "也可以从下面网址中直接下载 org.tensorflow.tensorflow的jar包\n",
    "\n",
    "以及其依赖的org.tensorflow.libtensorflow 和 org.tensorflowlibtensorflow_jni的jar包 放到项目中。\n",
    "\n",
    "https://mvnrepository.com/artifact/org.tensorflow/tensorflow/1.15.0"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 三、在spark(scala)项目中driver端加载tensorflow模型调试成功"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "我们的示范代码在jupyter notebook中进行演示，需要安装toree以支持spark(scala)。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```scala\n",
    "import scala.collection.mutable.WrappedArray\n",
    "import org.{tensorflow=>tf}\n",
    "\n",
    "//注：load函数的第二个参数一般都是“serve”，可以从模型文件相关信息中找到\n",
    "\n",
    "val bundle = tf.SavedModelBundle \n",
    "   .load(\"/Users/liangyun/CodeFiles/eat_tensorflow2_in_30_days/data/linear_model/1\",\"serve\")\n",
    "\n",
    "//注：在java版本的tensorflow中还是类似tensorflow1.0中静态计算图的模式，需要建立Session, 指定feed的数据和fetch的结果, 然后 run.\n",
    "//注：如果有多个数据需要喂入，可以连续用用多个feed方法\n",
    "//注：输入必须是float类型\n",
    "\n",
    "val sess = bundle.session()\n",
    "val x = tf.Tensor.create(Array(Array(1.0f,2.0f),Array(2.0f,3.0f)))\n",
    "val y =  sess.runner().feed(\"serving_default_inputs:0\", x)\n",
    "         .fetch(\"StatefulPartitionedCall:0\").run().get(0)\n",
    "\n",
    "val result = Array.ofDim[Float](y.shape()(0).toInt,y.shape()(1).toInt)\n",
    "y.copyTo(result)\n",
    "\n",
    "if(x != null) x.close()\n",
    "if(y != null) y.close()\n",
    "if(sess != null) sess.close()\n",
    "if(bundle != null) bundle.close()  \n",
    "\n",
    "result\n",
    "\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "输出如下：\n",
    "\n",
    "```\n",
    "Array(Array(3.019596), Array(3.9878292))\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![](./data/TfDriver.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 四、在spark(scala)项目中通过RDD在excutor上加载tensorflow模型调试成功"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "下面我们通过广播机制将Driver端加载的TensorFlow模型传递到各个excutor上，并在excutor上分布式地调用模型进行推断。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```scala\n",
    "import org.apache.spark.sql.SparkSession\n",
    "import scala.collection.mutable.WrappedArray\n",
    "import org.{tensorflow=>tf}\n",
    "\n",
    "val spark = SparkSession\n",
    "    .builder()\n",
    "    .appName(\"TfRDD\")\n",
    "    .enableHiveSupport()\n",
    "    .getOrCreate()\n",
    "\n",
    "val sc = spark.sparkContext\n",
    "\n",
    "//在Driver端加载模型\n",
    "val bundle = tf.SavedModelBundle \n",
    "   .load(\"/Users/liangyun/CodeFiles/master_tensorflow2_in_20_hours/data/linear_model/1\",\"serve\")\n",
    "\n",
    "//利用广播将模型发送到excutor上\n",
    "val broads = sc.broadcast(bundle)\n",
    "\n",
    "//构造数据集\n",
    "val rdd_data = sc.makeRDD(List(Array(1.0f,2.0f),Array(3.0f,5.0f),Array(6.0f,7.0f),Array(8.0f,3.0f)))\n",
    "\n",
    "//通过mapPartitions调用模型进行批量推断\n",
    "val rdd_result = rdd_data.mapPartitions(iter => {\n",
    "    \n",
    "    val arr = iter.toArray\n",
    "    val model = broads.value\n",
    "    val sess = model.session()\n",
    "    val x = tf.Tensor.create(arr)\n",
    "    val y =  sess.runner().feed(\"serving_default_inputs:0\", x)\n",
    "             .fetch(\"StatefulPartitionedCall:0\").run().get(0)\n",
    "\n",
    "    //将预测结果拷贝到相同shape的Float类型的Array中\n",
    "    val result = Array.ofDim[Float](y.shape()(0).toInt,y.shape()(1).toInt)\n",
    "    y.copyTo(result)\n",
    "    result.iterator\n",
    "    \n",
    "})\n",
    "\n",
    "\n",
    "rdd_result.take(5)\n",
    "bundle.close\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "输出如下：\n",
    "\n",
    "```\n",
    "Array(Array(3.019596), Array(3.9264367), Array(7.8607616), Array(15.974984))\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![](./data/TfRDD.png)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 五、在spark(scala)项目中通过DataFrame在excutor上加载tensorflow模型调试成功"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "除了可以在Spark的RDD数据上调用tensorflow模型进行分布式推断，\n",
    "\n",
    "我们也可以在DataFrame数据上调用tensorflow模型进行分布式推断。\n",
    "\n",
    "主要思路是将推断方法注册成为一个sparkSQL函数。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```scala\n",
    "import org.apache.spark.sql.SparkSession\n",
    "import scala.collection.mutable.WrappedArray\n",
    "import org.{tensorflow=>tf}\n",
    "\n",
    "object TfDataFrame extends Serializable{\n",
    "    \n",
    "    \n",
    "    def main(args:Array[String]):Unit = {\n",
    "        \n",
    "        val spark = SparkSession\n",
    "        .builder()\n",
    "        .appName(\"TfDataFrame\")\n",
    "        .enableHiveSupport()\n",
    "        .getOrCreate()\n",
    "        val sc = spark.sparkContext\n",
    "        \n",
    "        \n",
    "        import spark.implicits._\n",
    "\n",
    "        val bundle = tf.SavedModelBundle \n",
    "           .load(\"/Users/liangyun/CodeFiles/master_tensorflow2_in_20_hours/data/linear_model/1\",\"serve\")\n",
    "\n",
    "        val broads = sc.broadcast(bundle)\n",
    "        \n",
    "        //构造预测函数，并将其注册成sparkSQL的udf\n",
    "        val tfpredict = (features:WrappedArray[Float])  => {\n",
    "            val bund = broads.value\n",
    "            val sess = bund.session()\n",
    "            val x = tf.Tensor.create(Array(features.toArray))\n",
    "            val y =  sess.runner().feed(\"serving_default_inputs:0\", x)\n",
    "                     .fetch(\"StatefulPartitionedCall:0\").run().get(0)\n",
    "            val result = Array.ofDim[Float](y.shape()(0).toInt,y.shape()(1).toInt)\n",
    "            y.copyTo(result)\n",
    "            val y_pred = result(0)(0)\n",
    "            y_pred\n",
    "        }\n",
    "        spark.udf.register(\"tfpredict\",tfpredict)\n",
    "        \n",
    "        //构造DataFrame数据集，将features放到一列中\n",
    "        val dfdata = sc.parallelize(List(Array(1.0f,2.0f),Array(3.0f,5.0f),Array(7.0f,8.0f))).toDF(\"features\")\n",
    "        dfdata.show \n",
    "        \n",
    "        //调用sparkSQL预测函数，增加一个新的列作为y_preds\n",
    "        val dfresult = dfdata.selectExpr(\"features\",\"tfpredict(features) as y_preds\")\n",
    "        dfresult.show \n",
    "        bundle.close\n",
    "    }\n",
    "}\n",
    "\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![](./data/TfDataFrame.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "以上我们分别在spark 的RDD数据结构和DataFrame数据结构上实现了调用一个tf.keras实现的线性回归模型进行分布式模型推断。\n",
    "\n",
    "在本例基础上稍作修改则可以用spark调用训练好的各种复杂的神经网络模型进行分布式模型推断。\n",
    "\n",
    "但实际上tensorflow并不仅仅适合实现神经网络，其底层的计算图语言可以表达各种数值计算过程。\n",
    "\n",
    "利用其丰富的低阶API，我们可以在tensorflow2.0上实现任意机器学习模型，\n",
    "\n",
    "结合tf.Module提供的便捷的封装功能，我们可以将训练好的任意机器学习模型导出成模型文件并在spark上分布式调用执行。\n",
    "\n",
    "这无疑为我们的工程应用提供了巨大的想象空间。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
